package org.wj.accumulator

import org.apache.spark.rdd.RDD
import org.apache.spark.util.CollectionAccumulator
import org.apache.spark.{SparkConf, SparkContext}

object AccumulatorUserSex extends App {


  val conf = new SparkConf().setMaster("local").setAppName(this.getClass.getName)

  val sc = new SparkContext(conf)

  private val rdd: RDD[(String, String, Int)] = sc.parallelize(Array(("tom", "男", 15), ("tom", "男", 15), ("tom", "男", 15), ("tom", "男", 15)))

  //自定义累加器
  val accumulator:UserAccumulator[User] = new UserAccumulator[User]

  sc.register(accumulator, "性别统计")

  //长整数累加器
  val longAccumulator = sc.longAccumulator("人员统计")

  //集合累加器
  private val collectionAccumulator: CollectionAccumulator[String] = sc.collectionAccumulator[String]("姓名集合")

  rdd.map(t3 => User(t3._1,t3._2, t3._3)).foreach(u => {
    accumulator.add(u)
    longAccumulator.add(1)
    collectionAccumulator.add(u.name)
  })

  println(accumulator)

  println(longAccumulator)

  println(collectionAccumulator)

}
